package com.code.coinprice.tasks;

import com.alibaba.fastjson.JSON;
import com.code.coinprice.configs.NettyConfig;
import com.code.coinprice.consts.CommonConstant;
import com.code.coinprice.handlers.RedisHandler;
import com.code.coinprice.managers.CoinPriceManager;
import com.code.coinprice.models.CoinPriceSource;
import com.code.coinprice.models.RelativeCoinPrice;
import com.code.coinprice.services.CoinPriceDictService;
import com.code.coinprice.services.CoinPriceSourceService;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * 同步币价定时器
 *
 * @author xiaoyaowang
 */
@Component
public class SynCoinPriceSchedule {

    private static final Logger log = LoggerFactory.getLogger(SynCoinPriceSchedule.class);

    @Resource
    private CoinPriceSourceService coinPriceSourceService;

    @Resource
    private CoinPriceManager coinPriceManager;

    @Resource
    private RedisHandler redisHandler;

    @Resource
    private CoinPriceDictService coinPriceDictService;

    @Scheduled(cron = "*/1 * * * * ?")
    public void synCoinPrice() {
        String scheduleTaskOn = coinPriceDictService.getDictValueByDictKey(CommonConstant.SCHEDULE_TASK_ON);
        // 定时器一旦关闭就有可能触发netty的相关连接读写空闲
        if ("0".equals(scheduleTaskOn)) {
            return;
        }
        List<CoinPriceSource> coinPriceSourceList = coinPriceSourceService.getCoinPriceSourceList();
        if (!CollectionUtils.isEmpty(coinPriceSourceList)) {
            ConcurrentHashMap<String, Channel> platformChannelMap = NettyConfig.getPlatformChannelMap();
            String source;
            Channel channel;
            List<RelativeCoinPrice> relativeCoinPrices;
            Set<Object> sets;
            for (CoinPriceSource coinPriceSource : coinPriceSourceList) {
                source = coinPriceSource.getSource();
                sets = redisHandler.zgetRange(source, 0, -1);
                if (!CollectionUtils.isEmpty(sets)) {
                    for (Object clientChannelId : sets) {
                        channel = platformChannelMap.get(clientChannelId);
                        if (channel != null) {
                            relativeCoinPrices = coinPriceManager.getCoinPriceListBySource(source);
                            if (!CollectionUtils.isEmpty(relativeCoinPrices)) {
                                String coinType = (String) redisHandler.get(clientChannelId + ":" + CommonConstant.COIN_TYPE);
                                if (!StringUtils.isEmpty(coinType)) {
                                    relativeCoinPrices = relativeCoinPrices.stream().filter(item -> coinType.equalsIgnoreCase(item.getCoinName())).collect(Collectors.toList());
                                }
                                channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(relativeCoinPrices)));
                            }
                        }
                    }
                }

            }
        }

    }
}
